package org.protocol;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.Message;
import org.jetlinks.core.message.codec.*;
import org.jetlinks.core.message.function.FunctionInvokeMessage;
import org.jetlinks.core.message.property.ReportPropertyMessage;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;

import javax.annotation.Nonnull;

public class MqttDeviceMessageCodec implements DeviceMessageCodec {
    public Transport getSupportTransport() {
        return DefaultTransport.MQTT;
    }

    @Nonnull
    public Publisher<? extends Message> decode(@Nonnull MessageDecodeContext messageDecodeContext) {
        FromDeviceMessageContext ctx = ((FromDeviceMessageContext) messageDecodeContext);
        MqttMessage message = ((MqttMessage) ctx.getMessage());

        String topic = message.getTopic();
        JSONObject payload = message.payloadAsJson();

        //判断topic
        if (topic.equals("/properties/report")) {
            ReportPropertyMessage report = new ReportPropertyMessage();
            //整个请求内容作为属性
            report.properties(payload);
            return Flux.just(report);
        }
        return Flux.empty();
    }

    @Nonnull
    public Publisher<? extends EncodedMessage> encode(@Nonnull MessageEncodeContext messageEncodeContext) {
        ToDeviceMessageContext ctx = ((ToDeviceMessageContext) messageEncodeContext);

        DeviceMessage message = ((DeviceMessage) ctx.getMessage());

        if (message instanceof FunctionInvokeMessage) {
            FunctionInvokeMessage msg = ((FunctionInvokeMessage) message);

            return Flux.just(
                SimpleMqttMessage
                    .builder()
                    .topic("/invoke/" + msg.getFunctionId())
                    .payload(JSON.toJSONString(msg.inputsToMap()))
                    .build()
            );
        }

        return Flux.empty();
    }
}
